package com.lj.spring.hbase_hadoop;

import java.util.List;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.testng.AbstractTestNGSpringContextTests;
import org.springframework.transaction.annotation.Transactional;
import org.testng.Assert;
import org.testng.annotations.Test;

import com.lj.tools.PrintTool;

/**
 * @Transactional is use to make the method has transaction
 * 
 * Test is okay
 * @author root
 *
 */


@ContextConfiguration({"classpath:spring/applicationContext.xml"})
//@Transactional
public class HBaseSpringTestFrame extends AbstractTestNGSpringContextTests{
	//@Autowired :Default use class type to match
	@Autowired
	HbaseTemplate hbaseTemplate;
	//自动加载spring的context
	@Autowired
	private ApplicationContext ctx;
	
	//ApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:spring/applicationContext.xml");
	
	private String tableName = "test4";
	
	private String columnFamily = "a";
	
	@Test
	@Transactional
    public void testFind() {
		//search the default namespace data;
		System.out.println("hbase.zookeeper.quorum:"+hbaseTemplate.getConfiguration().get("hbase.zookeeper.quorum"));
		System.out.println("hbase.master:"+hbaseTemplate.getConfiguration().get("hbase.master"));
		System.out.println("hbase.zookeeper.property.clientPort:"+hbaseTemplate.getConfiguration().get("hbase.zookeeper.property.clientPort"));
		//hbaseTest.getConfiguration().set("hbase.master","172.16.135.160:60000");
		//System.out.println("hbase.master2:"+hbaseTest.getConfiguration().get("hbase.master"));
		List<String> rows = hbaseTemplate.find("test4", columnFamily, new RowMapper<String>() {

			//这里是对每一个row的处理，对每个row处理的结果，将会返回给队列中的String
			@Override
			public String mapRow(Result result, int rowNum) throws Exception {
				// TODO Auto-generated method stub
				String whatIwant="";
				//cell is use rowkey,family,qualifier
				for(Cell cell:result.rawCells()){
					//开销太大，不推荐使用
					//cell.getQualifier();
					String qulifier = new String(CellUtil.cloneQualifier(cell),"UTF-8");
					String value = new String(CellUtil.cloneValue(cell));
					//System.out.println("Get the value:"+value);
					whatIwant = whatIwant+qulifier+":"+value+" ";
					
				}
				return whatIwant;
			}
		});
		
		PrintTool.print(rows);
       
    }
	
	/**
	 * 测试成功
	 */
	@Test
	public void testExecute(){
		hbaseTemplate.execute(tableName, new TableCallback<Object>() {

			
			@Override
			public Object doInTable(HTableInterface table) throws Throwable {
				 Put p = new Put(Bytes.toBytes("2"));
				    p.addColumn(Bytes.toBytes("a"), Bytes.toBytes("测试中文列"), Bytes.toBytes("AValue"));
				    table.put(p);
				
				return null;
			}

		});
	}
	
	@Test
	public void testBean(){
		Assert.assertTrue(ctx.containsBean("hbaseTemplate"));
		
		
	}

}
